package study.mq.kafka.consumer.support.rebalance;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;

import java.util.Collection;

@Slf4j
public class CustomRebalanceListener implements ConsumerRebalanceListener {

    /**
     * 分区丢失
     *
     * @param partitions 分区信息
     */
    @Override
    public void onPartitionsLost(Collection<TopicPartition> partitions) {
        log.info(">>>>>>>>>> onPartitionsLost >>>>>>>>>>");
        partitions.forEach(tp -> {
            String topic = tp.topic();
            int partition = tp.partition();
            log.info("lost partitions= [{}, {}]", topic, partition);
        });
    }

    /**
     * 分区取消
     *
     * @param partitions 分区信息
     */
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.info(">>>>>>>>>> onPartitionsRevoked >>>>>>>>>>");
        partitions.forEach(tp -> {
            String topic = tp.topic();
            int partition = tp.partition();
            log.info("revoked partitions= [{}, {}]", topic, partition);
        });
    }

    /**
     * 分区分配
     *
     * @param partitions 分区信息
     */
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info(">>>>>>>>>> onPartitionsAssigned >>>>>>>>>>");

        partitions.forEach(tp -> {
            String topic = tp.topic();
            int partition = tp.partition();
            log.info("assigned partitions= [{}, {}]", topic, partition);
        });
    }
}
